package com.kingsoft.dc.khaos.module.mpp.operator;

import com.kingsoft.dc.khaos.module.mpp.client.ConfVars;
import com.kingsoft.dc.khaos.module.mpp.client.KhaosContants;
import com.kingsoft.dc.khaos.module.mpp.client.MppClient;
import com.kingsoft.dc.khaos.module.mpp.exception.MppSqlException;
import com.kingsoft.dc.khaos.module.mpp.metadata.CombineStrategy;
import com.kingsoft.dc.khaos.module.mpp.metadata.MppConfig;
import com.kingsoft.dc.khaos.module.mpp.model.ColumnEntity;
import com.kingsoft.dc.khaos.module.mpp.sqlbuilder.GetAddSqlBuilder;
import com.kingsoft.dc.khaos.module.mpp.sqlbuilder.SqlBuilderUtils;
import com.kingsoft.dc.khaos.plugin.core.KhaosContext;
import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import static org.apache.hadoop.yarn.webapp.hamlet.HamletSpec.Media.print;

/**
 * 拉链表.
 */
public class TyLink extends CombineStrategy {

    @Override
    public void execute(KhaosContext kc, String moduleId, MppConfig mppConfig) {
        //初始化MPP client,client已经初始化数据库连接
        MppClient client = null;
        String tmp_add_tname = "";
        String tmp_del_tname = "";
        String src_Type = "";
        try {
            client = new MppClient(mppConfig, kc);


            //获取操作的库表参数
            //比如获取源表所属的schema
            String srcSchema = String.format("\"%s\"", mppConfig.getSrcSchema());
            log().info("源库名srcSchema===>  " + srcSchema);
            //源表名
            String srcTable = String.format("\"%s\"", mppConfig.getSrcTable());
            log().info("源表名srcTable===>  " + srcTable);

            //源表过滤条件
            String filter = mppConfig.getDataFilter();

            //目标表所属schema名
            String tarSchema = String.format("\"%s\"", mppConfig.getTarSchema());
            log().info("目标库名tarSchema===>  " + tarSchema);
            //目标表名
            String tarTable = String.format("\"%s\"", mppConfig.getTarTable());
            log().info("目标表名tarTable===>  " + tarTable);


            //减档表所属schema名
            String delSchema = String.format("\"%s\"", mppConfig.getDelSchema());
            log().info("减档库名delSchema===>  " + delSchema);
            //减档表名
            String delTable = String.format("\"%s\"", mppConfig.getDelTable());
            log().info("减档表名delTable===>  " + delTable);

            //目标表名(上一业务日期)所属schema名
            //String preSchema = mppConfig.getPreSchema();

            //目标表名(上一业务日期)
            //String preTable = mppConfig.getPreTable();
            //主键列表
            List<String> keys = mppConfig.getpKeys();
            List<String> linkedKeys = new ArrayList<>();
            Iterator<String> it = mppConfig.getLinkedKeys().iterator();
            if (mppConfig.getLinkedKeys() != null && mppConfig.getLinkedKeys().size() > 0) {
                while (it.hasNext()) {
                    String lkey = (String.format("\"%s\"", it.next()));
                    linkedKeys.add(lkey);
                }
            } else {
                throw new Exception("拉链算法未指定对比字段");
            }
            log().info("指定对比字段===>  " + StringUtils.join(linkedKeys, ","));
            //获取源档类型
            src_Type = mppConfig.getSrcType();
            //获取主键字符串
            //String pKeys = SqlBuilderUtils.selKeys(tarTable,keys);
            //是否物理删除
            boolean isDeleted = mppConfig.isDeleted();
            //获取追加字段名
            //比如获取[数据日期]字段名
            String bizDateColName = client.getAppendColName(ConfVars.P9_DATA_DATE);

            //获取追加字段值，比如获取[数据日期]字段值
            String bizDate = kc.getConf().getString(KhaosContants.BIZ_DATE);
            String batchNo = kc.getConf().getString(KhaosContants.BATCH_NO);
            String jobName = kc.getConf().getString(KhaosContants.JOB_NAME);
            String jobId = kc.getConf().getString(KhaosContants.JOB_INSTANCE_ID);
            //获取表全字段信息如id,name
            List<String> columnEntities = client.getColumnNameList(mppConfig.getSrcSchema(), mppConfig.getSrcTable());
            log().info("源表全字段信息===>  " + StringUtils.join(columnEntities, ","));

            //获取表及表全字段信息
            List<ColumnEntity> tableColumns = client.getColumnInfo(mppConfig.getSrcSchema(), mppConfig.getSrcTable());

            List<ColumnEntity> linkedColumns = client.subColumnInfo(mppConfig.getSrcSchema(), mppConfig.getSrcTable(), linkedKeys);
            log().info("指定对比字段信息===>  " + StringUtils.join(linkedColumns, ","));

            //获取目标表主键字
            String columns = SqlBuilderUtils.selKeys(tarTable, columnEntities);
            log().info("目标表主键字段信息===>  " + columns);
            //获取两表关联主键SQL字符串
            String pKs = SqlBuilderUtils.joinKeys(srcTable, tarTable, keys);
            //获取两表关联字段空值处理SQL字符串
            //String columnNulls =SqlBuilderUtils.replaceNull(srcTable,tarTable,tableColumns);
            //获取表别名+表全字段信息如s.id,s.name
            //String selTabCols = SqlBuilderUtils.selTabCols(srcTable,columnEntities);
            //获取目标表别名及表全字段信息如s.id,s.name
            String selTabColst = SqlBuilderUtils.selTabCols(tarTable, columnEntities);
            log().info("目标表表别名及表全字段信息===>  " + selTabColst);
            //获取表别名+表全字段信息如s.id,s.name
            // String selTabColst1 = SqlBuilderUtils.selTabCols(tarTable + "1",columnEntities);
            String tyStartDate = client.getAppendColName(ConfVars.P9_START_DATE);
            String tyStartBatch = client.getAppendColName(ConfVars.P9_START_BATCH);
            String tyEndDate = client.getAppendColName(ConfVars.P9_END_DATE);
            //求增量表、减量表
            String history_table = "( SELECT * FROM \n" + tarSchema + "." + tarTable + "\n where " + tyEndDate + " = to_date('29991231','YYYYMMDD') )";
            tmp_add_tname = tarSchema + ".\"tmp_" + jobId + "_add\"";           //源增量表
            tmp_del_tname = tarSchema + ".\"tmp_" + jobId + "_del\"";           //源减量表
            String sql_add_record = "";
            String sql_del_record = "";
            if (src_Type.equalsIgnoreCase("ALL")) {
                sql_add_record = " ( " + GetAddSqlBuilder.getAddSelectSql(srcSchema + "." + srcTable, history_table, keys, linkedColumns, filter) + " ) ";
                sql_del_record = " ( " + GetAddSqlBuilder.getDelSelectSql(srcSchema + "." + srcTable, history_table, keys, filter) + " ) ";
            } else {
                tmp_add_tname = srcSchema + "." + srcTable;                     //源增量表
                tmp_del_tname = delSchema + "." + delTable;           //源减量表
            }

            //获取目标表过滤条件
            String countwhere_sql = String.format(" (" + tyStartDate + " > to_date('%s','YYYYMMDD'))\n" +
                    "         or (\n" +
                    "         " + tyStartDate + " = to_date('%s','YYYYMMDD') \n" +
                    "         and \n" +
                    "         to_number(" + tyStartBatch + ",'9999') > to_number('%s','9999') " +
                    "         ) limit 1;", bizDate, bizDate, batchNo);

//            System.out.println(countwhere_sql);

            //获取目标表是否存在开链日期大于或等于业务日期的记录数
            Boolean srcCount = client.hasRecord(tarSchema, tarTable, countwhere_sql);

            //拉链算法只允许重跑最新一批次的作业检测SQL
            if (srcCount == false) {
                //拉链算法只允许重跑最新一批次的作业检测SQL
                String rerun_sql = rerunSql(tarSchema, tarTable, bizDate, jobName, batchNo, columns, client);
//                System.out.println(rerun_sql);
                client.execInsert(rerun_sql);

                //拼接拉链表表全量档特殊算子SQL
                String exeu_sql = buildSqlall(src_Type, columns, sql_add_record, sql_del_record, srcTable, tmp_add_tname, tmp_del_tname);
//                System.out.println(exeu_sql);
                client.execInsert(exeu_sql);

                //拼接拉链表减量全量算子SQL
                String exe_sql = buildSql(src_Type, tarSchema, tarTable, srcTable, bizDate, jobName, batchNo, columns, pKs, selTabColst, isDeleted, tmp_del_tname, client);
//                System.out.println(exe_sql);
                client.execInsert(exe_sql);

                //全拼接拉链表量增量减量算子SQL
                String exe_sql_all = "";
                if (null != filter && !filter.equals("")) {
                    exe_sql_all = buildallSqlWithFilter(src_Type, tarSchema, tarTable, srcTable, bizDate, jobName, batchNo, columns, pKs, selTabColst, isDeleted, tmp_add_tname, client, filter);
                } else {
                    exe_sql_all = buildallSql(src_Type, tarSchema, tarTable, srcTable, bizDate, jobName, batchNo, columns, pKs, selTabColst, isDeleted, tmp_add_tname, client);
                }
                client.execInsert(exe_sql_all);
                //清理临时表
                if (src_Type.equalsIgnoreCase("ALL")) {
                    String truncate_drop_sql = deleteTmpTable(tmp_add_tname, tmp_del_tname);
                    client.execInsert(truncate_drop_sql);
                }

            } else {
                System.out.println("拉链算法只允许重跑最新一批次");
                throw new Exception("拉链算法只允许重跑最新一批次");
            }
        } catch (Exception e) {
            e.printStackTrace();
            //清理临时表
            if (src_Type.equalsIgnoreCase("ALL")) {
                String truncate_drop_sql = deleteTmpTable(tmp_add_tname, tmp_del_tname);
                client.execInsert(truncate_drop_sql);
            }
            throw new MppSqlException("执行MPP SQL 异常", e);
        } finally {
            if (client != null) {
                client.close();
            }
        }
    }

    //清理临时表
    public String deleteTmpTable(String tmp_add_tname, String tmp_del_tname) {
        String all_sql = "";
        String truncate_add_sql = String.format("truncate table %s; \n", tmp_add_tname);
        String drop_add_sql = String.format("drop table if exists  %s; \n", tmp_add_tname);
        String truncate_del_sql = String.format("truncate table %s; \n", tmp_del_tname);
        String drop_del_sql = String.format("drop table if exists  %s; \n", tmp_del_tname);
        all_sql = "BEGIN WORK; \n" + truncate_add_sql + drop_add_sql + truncate_del_sql + drop_del_sql + " COMMIT ; \n ";
        return all_sql;
    }

    //重跑作业清理
    public String rerunSql(String tarSchema,
                           String tarTable,
                           String bizDate,
                           String jobName,
                           String batchNo,
                           String columns,
                           MppClient client) {
        System.out.println("TyLink build rerunSql start...");
        String tyStartDate = client.getAppendColName(ConfVars.P9_START_DATE);
        String tyStartBatch = client.getAppendColName(ConfVars.P9_START_BATCH);
        String tyEndDate = client.getAppendColName(ConfVars.P9_END_DATE);
        String tyEndBatch = client.getAppendColName(ConfVars.P9_END_BATCH);
        String tyDelFlag = client.getAppendColName(ConfVars.P9_DEL_FLAG);
        String tyJobName = client.getAppendColName(ConfVars.P9_JOB_NAME);
        //重跑作业清理 第1步
        String del_sql_one = String.format("\n BEGIN WORK; \n" +
                "\n delete from %s.%s \n" +
                "\t where " + tyStartDate + " = to_date('%s','YYYYMMDD')\n" +
                "\t   and " + tyStartBatch + " = '%s'; \n", tarSchema, tarTable, bizDate, batchNo);
        //第2步
        String del_sql_two = String.format("\n insert into %s.%s (\n" +
                "\t      %s,\n" +
                "\t      " + tyStartDate + ",\n" +
                "\t      " + tyStartBatch + ",\n" +
                "\t      " + tyEndDate + ",\n" +
                "\t      " + tyEndBatch + ",\n" +
                "\t      " + tyDelFlag + ",\n" +
                "\t      " + tyJobName + " \n" +
                "\t) select \n" +
                "\t      %s,\n" +
                "\t      " + tyStartDate + ",\n" +
                "\t      " + tyStartBatch + ",\n" +
                "\t      to_date('29991231','YYYYMMDD') as " + tyEndDate + ",\n" +
                "\t      '0000' as " + tyEndBatch + ",\n" +
                "\t      '0'  as " + tyDelFlag + ",\n" +
                "\t      '%s' as " + tyJobName + " \n" +
                "\t  from %s.%s \n" +
                "\t  where  " + tyEndDate + " = to_date('%s','YYYYMMDD')\n" +
                "\t     and  " + tyEndBatch + " = '%s'; \n", tarSchema, tarTable, columns, columns, jobName, tarSchema, tarTable, bizDate, batchNo);

        //第3步
        String del_sql_three = String.format("\n delete from %s.%s \n" +
                "       where " + tyEndDate + " = to_date('%s','YYYYMMDD')\n" +
                "         and " + tyEndBatch + " = '%s';", tarSchema, tarTable, bizDate, batchNo);

        String del_sql = del_sql_one + del_sql_two + del_sql_three;
        System.out.println("TyLink build rerunSql end...");
        return del_sql;
    }

    //拼接拉链表表全量档算子SQL实现类
    public String buildSqlall(String src_Type, String columns, String sql_add_record, String sql_del_record, String srcTable, String tmp_add_tname, String tmp_del_tname) {
        System.out.println("TyLink buildSqlall start...");
        String all_sql = "";
        //全量档
        if (src_Type.equalsIgnoreCase("ALL")) {
            //全量求增量临时表
            String all_del_sql_one = String.format("\n drop table if exists %s;\n" +
                            "create table %s " +
                            "as %s limit 0;\n" +
                            "truncate table %s;\n" +
                            "insert into %s  ( %s ) " +
                            "select  %s  from  %s  %s;\n"
                    , tmp_add_tname, tmp_add_tname, sql_add_record, tmp_add_tname, tmp_add_tname, columns, columns, sql_add_record, srcTable);

            //全量求减量临时表
            String all_del_sql_two = String.format("\n drop table if exists %s;\n" +
                            "create table %s " +
                            "as %s  limit 0;\n" +
                            "truncate table %s;\n" +
                            "insert into %s ( %s ) " +
                            "select  %s  from  %s  %s ;\n"
                    , tmp_del_tname, tmp_del_tname, sql_del_record, tmp_del_tname, tmp_del_tname, columns, columns, sql_del_record, srcTable);

            all_sql = all_sql + all_del_sql_one + all_del_sql_two;
        }
        System.out.println("TyLink buildSqlall end...");
        return all_sql;
    }

    //拼接拉链表表减量全量档算子SQL实现类
    public String buildSql(String src_Type,
                           String tarSchema,
                           String tarTable,
                           String srcTable,
                           String bizDate,
                           String jobName,
                           String batchNo,
                           String columns,
                           String pKs,
                           String selTabColst,
                           boolean isDeleted,
                           String tmp_del_tname,
                           MppClient client) {
        System.out.println("TyLink buildSql start...");
        String tyStartDate = client.getAppendColName(ConfVars.P9_START_DATE);
        String tyStartBatch = client.getAppendColName(ConfVars.P9_START_BATCH);
        String tyEndDate = client.getAppendColName(ConfVars.P9_END_DATE);
        String tyEndBatch = client.getAppendColName(ConfVars.P9_END_BATCH);
        String tyDelFlag = client.getAppendColName(ConfVars.P9_DEL_FLAG);
        String tyJobName = client.getAppendColName(ConfVars.P9_JOB_NAME);
        String all_sql = "";
        //减量以及全量且有物理删除的需要闭链
        if (((src_Type.equalsIgnoreCase("ALL")) && isDeleted) | (src_Type.equalsIgnoreCase("ADD") && isDeleted)) {
            //减量以及全量且有物理删除的需要闭链
            String exe_sql_one = String.format("\n  insert into %s.%s (\n" +
                    "\t      %s,\n" +
                    "\t      " + tyStartDate + ",\n" +
                    "\t      " + tyStartBatch + ",\n" +
                    "\t      " + tyEndDate + ",\n" +
                    "\t      " + tyEndBatch + ",\n" +
                    "\t      " + tyDelFlag + ",\n" +
                    "\t      " + tyJobName + " \t\n" +
                    "\t) select \n" +
                    "\t      %s,\n" +
                    "\t      " + tyStartDate + ",\n" +
                    "\t      " + tyStartBatch + ",\n" +
                    "\t      to_date('%s','YYYYMMDD') as " + tyEndDate + ",\n" +
                    "\t      '%s' as " + tyEndBatch + ",\n" +
                    "\t      '1'  as " + tyDelFlag + ",\n" +
                    "\t      '%s' as " + tyJobName + " \n" +
                    "\t  from  (select * from %s.%s  where " + tyEndDate + " = to_date('29991231','YYYYMMDD')) %s \n" +
                    "\t  where exists ( select 1 from  %s   as %s  where %s );\n", tarSchema, tarTable, columns, selTabColst, bizDate, batchNo, jobName, tarSchema, tarTable, tarTable, tmp_del_tname, srcTable, pKs);

            String exe_sql_two = String.format("\n  delete from %s.%s %s using  %s  %s \n" +
                    "\t  where %s\n" +
                    "\t    and %s." + tyEndDate + " = to_date('29991231','YYYYMMDD');\n", tarSchema, tarTable, tarTable, tmp_del_tname, srcTable, pKs, tarTable);

            all_sql = all_sql + exe_sql_one + exe_sql_two;
        }
        System.out.println("TyLink buildSql end...");
        return all_sql;
    }

    //拼接拉链表表减量增量全量档算子SQL实现类
    public String buildallSql(String src_Type,
                              String tarSchema,
                              String tarTable,
                              String srcTable,
                              String bizDate,
                              String jobName,
                              String batchNo,
                              String columns,
                              String pKs,
                              String selTabColst,
                              boolean isDeleted,
                              String tmp_add_tname,
                              MppClient client) {
        System.out.println("TyLink buildallSql start...");
        String tyStartDate = client.getAppendColName(ConfVars.P9_START_DATE);
        String tyStartBatch = client.getAppendColName(ConfVars.P9_START_BATCH);
        String tyEndDate = client.getAppendColName(ConfVars.P9_END_DATE);
        String tyEndBatch = client.getAppendColName(ConfVars.P9_END_BATCH);
        String tyDelFlag = client.getAppendColName(ConfVars.P9_DEL_FLAG);
        String tyJobName = client.getAppendColName(ConfVars.P9_JOB_NAME);

        String all_sql = "";
        //1.增量或2.全量的增量部分3.增量配置减量的增量部分
        if ((src_Type.equalsIgnoreCase("ALL")) || (src_Type.equalsIgnoreCase("ADD")) || (src_Type.equalsIgnoreCase("ADD") && isDeleted)) {
            //1.更新记录需要闭链
            String exe_sql_one = String.format("insert into %s.%s (\n" +
                    "\t      %s,\n" +
                    "\t      " + tyStartDate + ",\n" +
                    "\t      " + tyStartBatch + ",\n" +
                    "\t      " + tyEndDate + ",\n" +
                    "\t      " + tyEndBatch + ",\n" +
                    "\t      " + tyDelFlag + ",\n" +
                    "\t      " + tyJobName + " \t\n" +
                    "\t) select \n" +
                    "\t      %s,\n" +
                    "\t      " + tyStartDate + ",\n" +
                    "\t      " + tyStartBatch + ",\n" +
                    "\t      to_date('%s','YYYYMMDD') as " + tyEndDate + ",\n" +
                    "\t      '%s' as " + tyEndBatch + ",\n" +
                    "\t       " + tyDelFlag + ",\n" +
                    "\t      '%s' as " + tyJobName + " \n" +
                    "\t  from  (select * from %s.%s  where " + tyEndDate + " = to_date('29991231','YYYYMMDD')) %s \n" +
                    "\t  where exists (  select 1 from  %s   as %s  where %s );\n", tarSchema, tarTable, columns, selTabColst, bizDate, batchNo, jobName, tarSchema, tarTable, tarTable, tmp_add_tname, srcTable, pKs);

            String exe_sql_two = String.format("delete from %s.%s %s using  %s  %s \n" +
                    "\t  where %s." + tyEndDate + " = to_date('29991231','YYYYMMDD')\n" +
                    "\t    and %s;\n", tarSchema, tarTable, tarTable, tmp_add_tname, srcTable, tarTable, pKs);

            //2.新增数据要开链
            String exe_sql_three = String.format("insert into %s.%s (\n" +
                    "\t      %s,\n" +
                    "\t      " + tyStartDate + ",\n" +
                    "\t      " + tyStartBatch + ",\n" +
                    "\t      " + tyEndDate + ",\n" +
                    "\t      " + tyEndBatch + ",\n" +
                    "\t      " + tyDelFlag + ",\n" +
                    "\t      " + tyJobName + " \t\n" +
                    "\t) select \n" +
                    "\t      %s,\n" +
                    "\t      to_date('%s','YYYYMMDD') as " + tyStartDate + ",\n" +
                    "\t      '%s' as " + tyStartBatch + ",\n" +
                    "\t      to_date('29991231','YYYYMMDD') as " + tyEndDate + ", \n" +
                    "\t      '0000' as " + tyEndBatch + ", \n" +
                    "\t      '0' as " + tyDelFlag + ", " +
                    "\t      '%s' as " + tyJobName + " \n" +
                    "\t  from  %s %s ;\n", tarSchema, tarTable, columns, columns, bizDate, batchNo, jobName, tmp_add_tname, srcTable);

            all_sql = all_sql + exe_sql_one + exe_sql_two + exe_sql_three + "\n COMMIT ; \n ";
        }
        System.out.println("TyLink buildallSql end...");
        return all_sql;
    }

    //拼接拉链表表减量增量全量档算子SQL实现类
    public String buildallSqlWithFilter(String src_Type,
                                        String tarSchema,
                                        String tarTable,
                                        String srcTable,
                                        String bizDate,
                                        String jobName,
                                        String batchNo,
                                        String columns,
                                        String pKs,
                                        String selTabColst,
                                        boolean isDeleted,
                                        String tmp_add_tname,
                                        MppClient client,
                                        String filter) {
        System.out.println("TyLink buildallSqlWithFilter start...");
        String tyStartDate = client.getAppendColName(ConfVars.P9_START_DATE);
        String tyStartBatch = client.getAppendColName(ConfVars.P9_START_BATCH);
        String tyEndDate = client.getAppendColName(ConfVars.P9_END_DATE);
        String tyEndBatch = client.getAppendColName(ConfVars.P9_END_BATCH);
        String tyDelFlag = client.getAppendColName(ConfVars.P9_DEL_FLAG);
        String tyJobName = client.getAppendColName(ConfVars.P9_JOB_NAME);
        String schemaTblFilter = String.format("(select * from %s where %s)", tmp_add_tname, filter);
        String all_sql = "";
        //1.增量或2.全量的增量部分3.增量配置减量的增量部分
        if ((src_Type.equalsIgnoreCase("ALL")) || (src_Type.equalsIgnoreCase("ADD")) || (src_Type.equalsIgnoreCase("ADD") && isDeleted)) {
            //1.更新记录需要闭链
            String exe_sql_one = String.format("insert into %s.%s (\n" +
                    "\t      %s,\n" +
                    "\t      " + tyStartDate + ",\n" +
                    "\t      " + tyStartBatch + ",\n" +
                    "\t      " + tyEndDate + ",\n" +
                    "\t      " + tyEndBatch + ",\n" +
                    "\t      " + tyDelFlag + ",\n" +
                    "\t      " + tyJobName + " \t\n" +
                    "\t) select \n" +
                    "\t      %s,\n" +
                    "\t      " + tyStartDate + ",\n" +
                    "\t      " + tyStartBatch + ",\n" +
                    "\t      to_date('%s','YYYYMMDD') as " + tyEndDate + ",\n" +
                    "\t      '%s' as " + tyEndBatch + ",\n" +
                    "\t       " + tyDelFlag + ",\n" +
                    "\t      '%s' as " + tyJobName + " \n" +
                    "\t  from  (select * from %s.%s  where " + tyEndDate + " = to_date('29991231','YYYYMMDD')) %s \n" +
                    "\t  where exists (  select 1 from  %s   as %s  where %s );\n", tarSchema, tarTable, columns, selTabColst, bizDate, batchNo, jobName, tarSchema, tarTable, tarTable, tmp_add_tname, srcTable, pKs);

            String exe_sql_two = String.format("delete from %s.%s %s using  %s  %s \n" +
                    "\t  where %s." + tyEndDate + " = to_date('29991231','YYYYMMDD')\n" +
                    "\t    and %s;\n", tarSchema, tarTable, tarTable, tmp_add_tname, srcTable, tarTable, pKs);

            //2.新增数据要开链
            String exe_sql_three = String.format("insert into %s.%s (\n" +
                    "\t      %s,\n" +
                    "\t      " + tyStartDate + ",\n" +
                    "\t      " + tyStartBatch + ",\n" +
                    "\t      " + tyEndDate + ",\n" +
                    "\t      " + tyEndBatch + ",\n" +
                    "\t      " + tyDelFlag + ",\n" +
                    "\t      " + tyJobName + " \t\n" +
                    "\t) select \n" +
                    "\t      %s,\n" +
                    "\t      to_date('%s','YYYYMMDD') as " + tyStartDate + ",\n" +
                    "\t      '%s' as " + tyStartBatch + ",\n" +
                    "\t      to_date('29991231','YYYYMMDD') as " + tyEndDate + ", \n" +
                    "\t      '0000' as " + tyEndBatch + ", \n" +
                    "\t      '0' as " + tyDelFlag + ", " +
                    "\t      '%s' as " + tyJobName + " \n" +
                    "\t  from %s %s;\n", tarSchema, tarTable, columns, columns, bizDate, batchNo, jobName, schemaTblFilter, srcTable);

            all_sql = all_sql + exe_sql_one + exe_sql_two + exe_sql_three + "\n COMMIT ; \n ";
        }
        System.out.println("TyLink buildallSqlWithFilter end...");
        return all_sql;
    }
}
